package com.orderpay_detect

import com.orderpay_detect.bean.{OrderEvent, ReceiptEvent}
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.co.CoProcessFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment, _}
import org.apache.flink.util.Collector

/**
  * @Description: TODO QQ1667847363
  * @author: xiao kun tai
  * @date:2021/12/4 20:37
  *                 双流实时对账
  */
object TxMatch {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) //定义事件时间语义

    //从文件中读取数据，并转换成样例类,提取时间戳生成watermark
    val receiptPath: String = "OrderPayDetect/src/main/resources/ReceiptLog.csv"
    val orderPath: String = "OrderPayDetect/src/main/resources/OrderLog.csv"
    //读取支付数据
    val receiptEventStream: KeyedStream[ReceiptEvent, String] = env
      .readTextFile(receiptPath).map(data => {
      val arr: Array[String] = data.split(",")
      ReceiptEvent(arr(0), arr(1), arr(2).toLong)
    }).assignAscendingTimestamps(_.timestamp * 1000L)
      .keyBy(_.txId)
    //读取订单数据
    val orderEventStream:KeyedStream[OrderEvent, String] = env
      .readTextFile(orderPath).map(data => {
      val arr: Array[String] = data.split(",")
      OrderEvent(arr(0).toLong, arr(1), arr(2), arr(3).toLong)
    })
      .filter(_.eventType == "pay")
      .assignAscendingTimestamps(_.timestamp * 1000L)
      .keyBy(_.txId)

    //合并两条流，进行处理
    val resultStream: DataStream[(OrderEvent, ReceiptEvent)] = orderEventStream.connect(receiptEventStream)
      .process(new TxPayMatchResult)




    resultStream.print("matched")
    resultStream.getSideOutput(new OutputTag[OrderEvent]("unmatched-pay")).print("unmatched pays")
    resultStream.getSideOutput(new OutputTag[ReceiptEvent]("unmatched-receipt")).print("unmatched receipts")

    env.execute("tx match job")

  }

  class TxPayMatchResult extends CoProcessFunction[OrderEvent, ReceiptEvent, (OrderEvent, ReceiptEvent)] {
    //定义状态，保存当前交易对应的订单支付事件和到账事件
    lazy val payEventState: ValueState[OrderEvent] = getRuntimeContext
      .getState(new ValueStateDescriptor[OrderEvent]("pay", classOf[OrderEvent]))

    lazy val receiptEventState: ValueState[ReceiptEvent] = getRuntimeContext
      .getState(new ValueStateDescriptor[ReceiptEvent]("receipt", classOf[ReceiptEvent]))

    //侧输出流标签
     val unmatchedPayEventOutputTag = new OutputTag[OrderEvent]("unmatched-pay")
     val unmatchedReceiptEventOutputTag = new OutputTag[ReceiptEvent]("unmatched-receipt")

    override def processElement1(pay: OrderEvent, context: CoProcessFunction[OrderEvent, ReceiptEvent, (OrderEvent, ReceiptEvent)]#Context, collector: Collector[(OrderEvent, ReceiptEvent)]): Unit = {
      //订单支付来了，要判断之前是否有到账事件
      val receipt = receiptEventState.value()
      if (receipt != null) {
        //如果已经有receipt，正常输出匹配，清空状态
        collector.collect((pay, receipt))
        receiptEventState.clear()
        payEventState.clear()
      } else {
        //如果还没来，注册定时器开始等待5秒
        context.timerService().registerEventTimeTimer(pay.timestamp * 1000L+5000L)
        payEventState.update(pay)
      }
    }

    override def processElement2(receipt: ReceiptEvent, context: CoProcessFunction[OrderEvent, ReceiptEvent, (OrderEvent, ReceiptEvent)]#Context, collector: Collector[(OrderEvent, ReceiptEvent)]): Unit = {
      //到账事件来了，要判断之前是否有pay事件
      val pay = payEventState.value()
      if (pay != null) {
        //如果已经有pay，正常输出匹配，清空状态
        collector.collect((pay, receipt))
        receiptEventState.clear()
        payEventState.clear()
      } else {
        //如果还没来，注册定时器开始等待3秒
        context.timerService().registerEventTimeTimer(receipt.timestamp * 1000L+3000L)
        receiptEventState.update(receipt)
      }
    }

    override def onTimer(timestamp: Long, ctx: CoProcessFunction[OrderEvent, ReceiptEvent, (OrderEvent, ReceiptEvent)]#OnTimerContext, out: Collector[(OrderEvent, ReceiptEvent)]): Unit = {
        //定时器触发，判断状态中哪个还存在，就代表另一个没来,输出到侧输出流
      if(payEventState.value()!=null){
        ctx.output(unmatchedPayEventOutputTag,payEventState.value())
      }
      if(receiptEventState.value()!=null){
        ctx.output(unmatchedReceiptEventOutputTag,receiptEventState.value())
      }
      //清空状态
      payEventState.clear()
      receiptEventState.clear()
    }
  }

}
